/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.indexer.hyracks;

import java.nio.ByteBuffer;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.columns.DeltaEncodedAscendingUnsignedLongColumnFactory;
import edu.ucsd.db.bassarisk.storage.common.Cabinet;
import edu.ucsd.db.bassarisk.storage.loading.CabinetLoader;
import edu.ucsd.db.bassarisk.storage.loading.ColumnLoader;

public class CabinetLoadOperatorDescriptor extends
    AbstractSingleActivityOperatorDescriptor {
  private static final long serialVersionUID = 1L;
  private static final Logger LOG = Logger
      .getLogger(CabinetLoadOperatorDescriptor.class.getName());

  private final String cabinetFileName;

  public CabinetLoadOperatorDescriptor(final JobSpecification spec,
                                       final String cabinetFileName) {
    super(spec, 1, 0);
    this.cabinetFileName = cabinetFileName;
  }

  @Override
  public IOperatorNodePushable
      createPushRuntime(final IHyracksTaskContext ctx,
                        final IRecordDescriptorProvider recordDescProvider,
                        final int partition, final int nPartitions)
          throws HyracksDataException {
    LOG.info("Creating " + this.getClass().getName());
    // Prepare objects used by returned operator
    final FrameDeserializer deserializer =
        new FrameDeserializer(ctx.getFrameSize(),
                              HyracksPostingSerialization
                                  .positionedPostingsDescriptor());

    return new AbstractUnaryInputSinkOperatorNodePushable() {
      private final Logger LOG = Logger.getLogger(this.getClass().getName());

      private Cabinet index;
      private CabinetLoader indexLoader;
      private ColumnLoader<LongType> columnLoader;

      private String currentKey = "";

      @Override
      public void close() throws HyracksDataException {
        this.LOG.info("Closing: " + this.getClass().getName());
        if (this.columnLoader != null) {
          this.indexLoader.closeColumn(this.columnLoader);
          this.columnLoader = null;
        }
        this.index.close();
      }

      @Override
      public void fail() throws HyracksDataException {
        this.LOG.info("Closing due to failure: " + this.getClass().getName());
        if (this.columnLoader != null) {
          this.indexLoader.closeColumn(this.columnLoader);
          this.columnLoader = null;
        }
        this.index.close();
      }

      @Override
      public void nextFrame(final ByteBuffer buffer) throws HyracksDataException {
        this.LOG.info("IndexBuilder next frame.");
        ObjectLongReference appendValue = LongTools.INSTANCE.newReference();
        deserializer.reset(buffer);

        // Iterate over tuples in buffer, and insert them into inverted index
        while (!deserializer.done()) {
          Object[] record = deserializer.deserializeRecord();

          assert record.length == 3;
          String groupKey = (String) record[0];
          String tokenKey = (String) record[1];
          String value = (String) record[2];
          Integer position = (Integer) record[3];
          String key = groupKey + "#" + tokenKey;
          if (key.compareTo(this.currentKey) != 0) {
            // require that postings are sorted by key
            assert key.compareTo(this.currentKey) < 0;
            if (this.columnLoader != null) {
              int address = this.indexLoader.closeColumn(this.columnLoader);
// LOG.info("Closed column " + this.currentKey + " at address " + address + ".");
            }
            this.columnLoader =
                this.indexLoader
                    .startColumn(key,
                                 new DeltaEncodedAscendingUnsignedLongColumnFactory());
            this.currentKey = key;
          }

          long psn = position.longValue();

// LOG.info(String.format("Loading posting: g=%s, t=%s, v=%s, p=%s", 
//                        groupKey, tokenKey, value, position));
          appendValue.setValue(psn);
          this.columnLoader.appendBufferedValue(appendValue);
        }
      }

      @Override
      public void open() throws HyracksDataException {
        this.LOG.info("opening cabinet loader operator");

        this.index =
            new Cabinet(ctx, CabinetLoadOperatorDescriptor.this.cabinetFileName);
        this.indexLoader = this.index.makeLoader();
      }
    };
  }
}
